/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.clients.producer.internals;

import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.Rate;
import org.apache.kafka.common.utils.Time;


/**
 * 内存池
 */
public final class BufferPool {
    /**
     * 总内存
     */
    private final long totalMemory;
    /**
     * 代表的是一个批次的大小，默认情况一下，一个批次的大小是16K。
     */
    private final int poolableSize;
    /**
     * 重入锁
     */
    private final ReentrantLock lock;
      /**
     * 池子就是一个队列，队列里面放的就是一个块一块的内存,就是跟一个连接池是一个道理。
     */
    private final Deque<ByteBuffer> free;
    private final Deque<Condition> waiters;
    /**
     * 可用内存
     */
    private long availableMemory;
    private final Metrics metrics;
    private final Time time;
    private final Sensor waitTime;

    /**
     * 创建一个缓冲池
     * 
     * @param memory    缓冲池内存的最大值
     * @param poolableSize    要缓存在空闲列表中而不是释放有缓冲区的大小
     * @param metrics instance of Metrics
     * @param time time instance
     * @param metricGrpName logical group name for metrics
     */
    public BufferPool(long memory, int poolableSize, Metrics metrics, Time time, String metricGrpName) {
        this.poolableSize = poolableSize;
        this.lock = new ReentrantLock();
        this.free = new ArrayDeque<ByteBuffer>();
        this.waiters = new ArrayDeque<Condition>();
        this.totalMemory = memory;
        this.availableMemory = memory;
        this.metrics = metrics;
        this.time = time;
        this.waitTime = this.metrics.sensor("bufferpool-wait-time");
        MetricName metricName = metrics.metricName("bufferpool-wait-ratio",
                                                   metricGrpName,
                                                   "The fraction of time an appender waits for space allocation.");
        this.waitTime.add(metricName, new Rate(TimeUnit.NANOSECONDS));
    }

    /**
     * 重新分配制定大小的缓冲区,如果内存和缓冲区池不足,此方法将阻塞,配置为阻塞的模式
     *
     * 
     * @param size  要分配的缓冲池大小,以字节为单位
     * @param maxTimeToBlockMs 阻塞缓冲区内存最大的时间
     * @return The buffer   缓冲区字节为单位
     * @throws InterruptedException 如果线程在阻塞时被中断
     * @throws IllegalArgumentException 如果 大小大于缓冲区的总大小会补阻止
     */
    public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException {
        /**
         * 如果你想要申请的内存的大小，如果超过32M
         */
        if (size > this.totalMemory)
            throw new IllegalArgumentException("Attempt to allocate " + size
                                               + " bytes, but there is a hard limit of "
                                               + this.totalMemory
                                               + " on memory allocations.");
        /**
         * 加锁的代码
         */

        this.lock.lock();
        try {
            /**
             * poolableSize 代表的是一个批次的大小，默认情况一下，一个批次的大小是16K。
             *  如果我们这次申请的批次的大小等于 我们设定好的一个批次的大小
             *  并且我们的内存池不为空，那么直接从内存池里面获取一个块内存就可以使用了。
             *  我们场景驱动的方式，第一次进来内存池里面里面是没有内存的，所以这儿是获取不到内存。
             */

            if (size == poolableSize && !this.free.isEmpty())
                //弹出队列头部元素,队列为空时返回null
                return this.free.pollFirst();

            /**
             * 内存的个数 *  批次的大小 = free的大小
             * size: 我们这次要申请的内存
             * this.availableMemory + freeListSize 目前可用的总内存
             *
             */
            int freeListSize = this.free.size() * this.poolableSize;
            /**
             * this.availableMemory + freeListSize 目前可用的总内存 大于你要申请的内存。
             */
            if (this.availableMemory + freeListSize >= size) {

                /**
                 * 内存充足,进行内存扣减
                 */
                freeUp(size); //
                this.availableMemory -= size;
                lock.unlock();
                return ByteBuffer.allocate(size);
            } else {
                /**
                 * 还有一种情况就是，我们整个内存池 还剩10k的内存，但是我们这次申请的内存是32k
                 * 批次可能就是16k,但是我们的一条消息，就是32K -> max(16,32) = 当前批次 = 32K
                 * 如果超过了可用的内存大小,将会被阻塞,统计分配的内存
                 */
                int accumulated = 0;
                ByteBuffer buffer = null;
                Condition moreMemory = this.lock.newCondition();
                long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);   //等待被人释放内存

                this.waiters.addLast(moreMemory);
                /**
                 * 总的分配的思路，可能一下子分配不了这么大的内存，但是可以先有点分配一点。
                 * 如果分配的内存的大小 还是没有要申请的内存大小大,内存池就会一直分配的内存，等着别人会释放内存,一点一点的去分配。
                 *
                 */
                while (accumulated < size) {
                    long startWaitNs = time.nanoseconds();
                    long timeNs;
                    boolean waitingTimeElapsed;
                    try {

                        /**
                         * 在等待，等待别人释放内存。如果这儿的代码是等待wait操作,那么我们可以猜想一下，当有人释放内存的时候
                         * 肯定是得唤醒这儿的代码
                         * 假设，突然有人 往内存池里面还了内存。那么这儿的代码就可以被唤醒了。代码就继续往下执行。
                         */
                        waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
                    } catch (InterruptedException e) {
                        this.waiters.remove(moreMemory);
                        throw e;
                    } finally {
                        long endWaitNs = time.nanoseconds();
                        timeNs = Math.max(0L, endWaitNs - startWaitNs);
                        this.waitTime.record(timeNs, time.milliseconds());
                    }

                    if (waitingTimeElapsed) {
                        this.waiters.remove(moreMemory);
                        throw new TimeoutException("Failed to allocate memory within the configured max blocking time " + maxTimeToBlockMs + " ms.");
                    }

                    remainingTimeToBlockNs -= timeNs;
                                      /**
                     * 再次方式看一下，内存池里面有没有数据了。
                     * 如果内存池里面有数据,并且你申请的内存的大小就是一个批次的大小
                     */
                    if (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {

                        /**
                         * 这儿就可以直接获取到内存。
                         */
                        buffer = this.free.pollFirst();
                        accumulated = size;
                    } else {
                        /**
                         * 确保释放的内存达到请求的内存
                         */
                        freeUp(size - accumulated);
                        /**
                         * 可以给你分配的内存
                         */
                        int got = (int) Math.min(size - accumulated, this.availableMemory);
                        /**
                         * 做内存扣减
                         */
                        this.availableMemory -= got;
                        /**
                         * 累加已经分配了多少内存。
                         */
                        accumulated += got;
                    }
                }

                // remove the condition for this thread to let the next thread
                // in line start getting memory
                Condition removed = this.waiters.removeFirst();
                if (removed != moreMemory)
                    throw new IllegalStateException("Wrong condition: this shouldn't happen.");

                // signal any additional waiters if there is more memory left
                // over for them
                if (this.availableMemory > 0 || !this.free.isEmpty()) {
                    if (!this.waiters.isEmpty())
                        this.waiters.peekFirst().signal();
                }

                /**
                 * 解锁,归还Buffer
                 */
                lock.unlock();
                if (buffer == null)
                    return ByteBuffer.allocate(size);
                else
                    return buffer;
            }
        } finally {
            if (lock.isHeldByCurrentThread())
                lock.unlock();
        }
    }

    /**
     *
     * 尝试确保通过释放缓冲区分配的内存至少达到请求的字节数
     *
     */
    private void freeUp(int size) {
        while (!this.free.isEmpty() && this.availableMemory < size)
            this.availableMemory += this.free.pollLast().capacity();
    }

    /**
     * 回收buffer
     * 
     * @param buffer 要返回的buffer
     * @param size 标记为归还的buffer 的大小,注意:这可能小于缓冲区大小,因为缓冲区可能在压缩期间重新分配
     */
    public void deallocate(ByteBuffer buffer, int size) {
        lock.lock();
        try {
            //如果你还回来的内存的大小 就等于一个批次的大小，
            //我们的参数设置的内存是16K，你计算出来一个批次的大小也是16，申请的内存也是16k
            if (size == this.poolableSize && size == buffer.capacity()) {
                //内存里面的东西清空
                buffer.clear();
                //把内存放入到内存池
                this.free.add(buffer);
            } else {
                //但是如果 我们释放的内存的大小
                //不是一个批次的大小，那就把归为可用内存
                //等着垃圾回收即可
                this.availableMemory += size;
            }
            Condition moreMem = this.waiters.peekFirst();
            if (moreMem != null)
                //释放了内存（或者是还了内存以后）
                //都会唤醒等待内存的线程。

                //接下来是不是还是要唤醒正在等待分配内存的线程。
                moreMem.signal();
        } finally {
            lock.unlock();
        }
    }

    public void deallocate(ByteBuffer buffer) {
        deallocate(buffer, buffer.capacity());
    }

    /**
     * the total free memory both unallocated and in the free list
     */
    public long availableMemory() {
        lock.lock();
        try {
            return this.availableMemory + this.free.size() * this.poolableSize;
        } finally {
            lock.unlock();
        }
    }

    /**
     * Get the unallocated memory (not in the free list or in use)
     */
    public long unallocatedMemory() {
        lock.lock();
        try {
            return this.availableMemory;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 等待内存中的阻塞线程数
     * The number of threads blocked waiting on memory
     */
    public int queued() {
        lock.lock();
        try {
            return this.waiters.size();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 使用后,将被保持在先内存的Buffer大小
     *
     */
    public int poolableSize() {
        return this.poolableSize;
    }

    /**
     *
     * 这个池要管理的总区的线程大小
     */
    public long totalMemory() {
        return this.totalMemory;
    }


    Deque<Condition> waiters() {
        return this.waiters;
    }
}
